package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class CreateTableNode extends BaseSparkNode {


    @Override
    public SparkResult process(SparkParam sparkParam) {
        JSONObject expression = sparkParam.getNodeExpression();
        String tableName = expression.getString("tableName");
        Boolean incrementFlag = expression.getBoolean("incrementFlag");

        String url = expression.getString("url");
        String username = expression.getString("username");
        String password = expression.getString("password");
        Dataset<Row> dataset = sparkParam.getDatasetList().get(0);
        // 将数据写入数据库
        SaveMode saveMode = SaveMode.Overwrite;

        if (incrementFlag) {
            saveMode = SaveMode.Append;
        }
        dataset.write()
                .mode(saveMode)
                .jdbc(url, tableName, new java.util.Properties() {
                    {
                        setProperty("user", username);
                        setProperty("password", password);
                    }
                });
        return SparkResult.success(dataset);
    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.CREATE_TABLE;
    }
}
